package cn.caplike.demo.project.reactor.quickstart;

import org.junit.Test;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Description: Handling Errors<br>
 * Reference: <a href="https://projectreactor.io/docs/core/release/reference/#error.handling">Handling Errors</a>
 *
 * @author LiKe
 * @version 1.0.0
 * @date 2020-08-24 10:46
 */
public class HandlingErrors {

    // ~ 4.6.1 Error Handling Operators
    //   https://projectreactor.io/docs/core/release/reference/#_error_handling_operators

    // ~ Demonstrating the Terminal Aspect of onError
    //   https://projectreactor.io/docs/core/release/reference/#_demonstrating_the_terminal_aspect_of_onerror

    @Test
    public void testErrorReturn() throws InterruptedException {
        Flux<String> flux =
                Flux.interval(Duration.ofMillis(250))
                        .map(input -> {
                            if (input < 3) return "tick " + input;
                            throw new RuntimeException("boom");
                        })
                        .onErrorReturn("Uh oh");

        flux.subscribe(System.out::println);

        // Note that interval executes on a timer Scheduler by default.
        // If we want to run that example in a main class, we would need to add a sleep call here so that the application does not exit immediately without any value being produced.
        Thread.sleep(2100);
    }

    // ~ Retry
    //   https://projectreactor.io/docs/core/release/reference/#_retrying

    @Test
    public void testRetry() throws InterruptedException {
        Flux.interval(Duration.ofMillis(250))
                .map(input -> {
                    if (input < 3) return "tick " + input;
                    throw new RuntimeException("boom");
                })
                // Restarting tick from 0
                .retry(3)
                .elapsed()
                .subscribe(System.out::println, System.err::println);

        // Ensure we have enough time for our 4x2 ticks: 否则主线程都执行完了, Scheduler 还没执行, 这样就没法看到打印出来的值了.
        Thread.sleep(10100);
    }

    /**
     * Description : There is a more advanced version of retry (called retryWhen) that uses a “companion” Flux to tell whether or not a particular failure should retry.
     * This companion Flux is created by the operator but decorated by the user, in order to customize the retry condition.<br>
     *
     * <p>
     * Retry cycles go as follows:<br>
     * 1. Each time an error happens (giving potential for a retry), a RetrySignal is emitted into the companion Flux,
     * which has been decorated by your function. Having a Flux here gives a bird eye’s view of all the attempts so far.
     * The RetrySignal gives access to the error as well as metadata around it.<br>
     * 2. If the companion Flux emits a value, a retry happens.<br>
     * 3. If the companion Flux completes, the error is swallowed, the retry cycle stops, and the resulting sequence completes, too.<br>
     * 4. If the companion Flux produces an error (e), the retry cycle stops and the resulting sequence errors with e.
     * </p>
     */
    @Test
    public void testCompanion1() {
        Flux<String> flux = Flux
                // This continuously produces errors, calling for retry attempts.
                .<String>error(new IllegalArgumentException())
                // doOnError before the retry lets us log and see all failures.
                .doOnError(System.out::println)
                .retryWhen(Retry.from(companion -> {
                    // Here, we consider the first three errors as retry-able (take(3)) and then give up.
                    return companion.take(3);
                }));
        flux.subscribe(System.out::println);
    }

    @Test
    public void testCompanion2() {
        AtomicInteger errorCount = new AtomicInteger();
        Flux<String> flux =
                Flux.<String>error(new IllegalArgumentException())
                        .doOnError(e -> errorCount.incrementAndGet())
                        .retryWhen(Retry.from(companion ->
                                // We customize Retry by adapting from a Function lambda rather than providing a concrete class
                                companion.map(retrySignal -> {
                                    // The companion emits RetrySignal objects, which bear number of retries so far and last failure
                                    if (retrySignal.totalRetries() < 3)
                                        // To allow for three retries, we consider indexes < 3 and return a value to emit (here we simply return the index).
                                        return retrySignal.totalRetries();
                                    else
                                        // In order to terminate the sequence in error, we throw the original exception after these three retries.
                                        throw Exceptions.propagate(retrySignal.failure());
                                })
                        ));
        flux.subscribe(System.out::println);
        System.out.println(errorCount.get()); // 4
    }

    // ~ 4.6.2. Handling Exceptions in Operators or Functions
    //   https://projectreactor.io/docs/core/release/reference/#_handling_exceptions_in_operators_or_functions

    // As a rule of thumb, an unchecked exception is always propagated through onError.
    // For instance, throwing a RuntimeException inside a map function translates to an onError event, as the following code shows:

    @Test
    public void testHandlingExceptionInOperatorOrFunction() {
        Flux.just("foo")
                .map(s -> { throw new IllegalArgumentException(s); })
                .subscribe(v -> System.out.println("GOT VALUE"),
                        e -> System.err.println("ERROR: " + e));
    }

}
